package com.example.toy.core.table;

import com.example.toy.ToySystem;
import com.example.toy.core.ToySystemFactoryProducer;
import com.example.toy.core.commit.CommitMsg;
import com.example.toy.core.commit.CommitState;
import com.example.toy.core.commit.CommitType;
import com.example.toy.core.commit.ToyCommitAbstractFactory;
import com.example.toy.core.conf.ToyTableConfig;
import com.example.toy.core.fs.FsOfflineToyTableFileView;
import com.example.toy.core.partition.PartitionPathStyle;
import com.example.toy.core.record.key.generator.KeyGenerator;
import com.example.toy.core.utils.ToyPathDefine;
import com.example.toy.core.version.ToyVersion;
import java.util.HashMap;
import java.util.List;
import java.util.stream.Collectors;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * FsOfflineTableMeta
 *
 * @author shenbing
 * @date 2021/12/21
 */
@Getter
@Slf4j
public class FsOfflineTableMeta extends ToyTableMeta {

  private final FileSystem fs;
  private final TableIdentifier tableIdentifier;

  @SneakyThrows
  public FsOfflineTableMeta(ToyTable table, Path basePath, ToyVersion version) {
    super(table, basePath, version);
    this.fs = basePath.getFileSystem(new Configuration());
    this.tableConfig = ToyTableConfig.fromPath(ToyPathDefine.getTableConfigFilePath(basePath));
    this.tableIdentifier = toyTable.getTableIdentifier();

    ToySystem toySystem = this.toyTable.getToySystem();
    ToySystemFactoryProducer factory = toySystem.getSystemFactoryProducer();
    ToyCommitAbstractFactory commitFactory = factory.buildCommitFactory();
    this.commitView = commitFactory.getCommitView(this.tableIdentifier, basePath);
    this.commitMsg =
        this.commitView
            .getCommit(version)
            .orElseThrow(() -> new IllegalStateException("table version not exists: " + version));
    this.realtimeFileView =
        new FsOfflineToyTableFileView(getTableIdentifier(), getBasePath(), getDataDir());
    loadSchema();
    loadPartitionMap();
  }

  /** 加载所有的分区列表缓存到内存中 */
  private void loadPartitionMap() {
    this.partitionPathMap = new HashMap<>();
    List<FileStatus> partitionDirs = realtimeFileView.getPartitionDirs();
    PartitionPathStyle partitionPathStyle = this.getPartitionPathStyle();
    KeyGenerator partitionKeyGenerator = this.getTableConfig().getPartitionKeyGenerator();
    for (FileStatus partitionDir : partitionDirs) {
      List<Object> partitionParams =
          partitionPathStyle.fromPartitionPath(getPartitionSchema(), partitionDir.getPath());
      String partitionKey =
          partitionKeyGenerator.generate(getPartitionSchema(), partitionParams.toArray());
      this.partitionPathMap.put(partitionKey, partitionDir.getPath());
    }
  }

  /** 加载表结构 */
  private void loadSchema() {
    List<CommitMsg> alterColumns =
        this.commitView.getCommits(
            new CommitType[] {CommitType.ALTER_COLUMN},
            CommitState.COMPLETED_COMMIT_STATE,
            version.bigger(),
            null,
            null,
            null);

    if (alterColumns.isEmpty()) {
      // 如果在当前版本之后，没有修改过表字段，那么直接从TableConfig中加载Schema
      this.schema = tableConfig.getSchema();
    } else {
      this.schema = this.commitView.getTableSchema(version);
    }

    List<String> partitionFields = this.tableConfig.getPartitionFields();
    if (partitionFields.isEmpty()) {
      this.partitionSchema = Schema.createRecord("partition", null, null, false);
    } else {
      List<Schema.Field> fields =
          partitionFields.stream()
              .map(name -> this.schema.getField(name))
              .collect(Collectors.toList());
      this.partitionSchema = Schema.createRecord("partition", null, null, false, fields);
    }

    List<String> recordKeyFields = this.tableConfig.getRecordKeyFields();
    List<Schema.Field> keyFields =
        recordKeyFields.stream()
            .map(name -> this.schema.getField(name))
            .collect(Collectors.toList());
    this.keySchema = Schema.createRecord("record_key", null, null, false, keyFields);
  }
}
